package com.gitee.cui.zookeeper.publishsubscribe;

import com.gitee.cui.zk.ZKclient;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

/**
 * zookeeper的watcher演示
 *
 * @author cuiqingqiang
 * @version 1.0
 * @date 2021-11-06 19:49
 **/
@Slf4j
public class ZKWatcher {

	private String workerPath = "/test/listener/remoteNode";

	private String subWorkerPath = "/test/listener/remoteNode/id-";

	/**
	 * 通过watcher实现监听事件
	 */
	public void watcher() {
		ZKclient instance = ZKclient.instance;
		CuratorFramework client = instance.getClient();
		try {
			// 检查节点是否存在，如果不存在就创建
			boolean nodeExist = instance.isNodeExist(workerPath);
			if (!nodeExist) {
				instance.createNode(workerPath, null);
			}

			Watcher watcher = new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					log.info("监听到事件变化：{}", event);
				}
			};

			byte[] payload = client.getData().usingWatcher(watcher).forPath(workerPath);
			log.info("获取到的节点信息：{}", new String(payload));

			// 模拟改变两次节点数据
			client.setData().forPath(workerPath, "第一次改变节点信息".getBytes());
			client.setData().forPath(workerPath, "第二次改变节点信息".getBytes());

			Thread.sleep(Integer.MAX_VALUE);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 通过NodeCache方式实现监听事件
	 */
	public void nodeCache() {
		ZKclient instance = ZKclient.instance;
		CuratorFramework client = instance.getClient();
		try {
			// 判断节点是否存在，如果不存在则创建节点
			boolean nodeExist = instance.isNodeExist(workerPath);
			if (!nodeExist) {
				instance.createNode(workerPath, null);
			}

			// 构建NodeCache实例
			NodeCache nodeCache = new NodeCache(client, workerPath, false);
			// 创建NodeCache监听处理
			NodeCacheListener listener = new NodeCacheListener() {
				/**
				 * Called when a change has occurred
				 */
				@Override
				public void nodeChanged() throws Exception {
					ChildData data = nodeCache.getCurrentData();
					log.info("Znode节点状态该表，path = {}", nodeCache.getPath());
					log.info("Znode节点状态该表，data = {}", new String(data.getData()));
					log.info("Znode节点状态该表，state = {}", data.getStat());
				}
			};

			// 添加监听事件
			nodeCache.getListenable().addListener(listener);
			// 启动缓存
			nodeCache.start();

			// 更改节点数据
			client.setData().forPath(workerPath, "第一次更改节点数据".getBytes());
			Thread.sleep(1000);
			client.setData().forPath(workerPath, "第二次更改节点数据".getBytes());
			Thread.sleep(1000);
			client.setData().forPath(workerPath, "第三次更改节点数据".getBytes());

			Thread.sleep(Integer.MAX_VALUE);
		} catch (Exception e) {
			log.error("创建NodeCache节点监听失败，workerPath = {}", workerPath);
		}
	}

	/**
	 * 通过PathChildrenCache实现监听事件
	 */
	public void pathChildrenCache() {
		ZKclient instance = ZKclient.instance;
		CuratorFramework client = instance.getClient();
		try {
			boolean nodeExist = instance.isNodeExist(workerPath);
			if (!nodeExist) {
				instance.createNode(workerPath, null);
			}

			// 创建监听事件
			PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
				@Override
				public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
					ChildData data = event.getData();
					String path = data.getPath();
					byte[] bytes = data.getData();
					switch (event.getType()) {
						case CHILD_ADDED:
							// 子节点添加事件
							log.info("子节点增加，path = {}, data = {}", path, new String(bytes));
							break;
						case CHILD_UPDATED:
							// 子节点更新
							log.info("子节点更新，path = {}, data = {}", path, new String(bytes));
							break;
						case CHILD_REMOVED:
							// 子节点删除
							log.info("子节点删除，path = {}, data = {}", path, new String(bytes));
							break;
						default:
							break;
					}
				}
			};

			// 构建PathChildrenCache实力
			PathChildrenCache pathChildrenCache = new PathChildrenCache(client, workerPath, true);

			// 添加事件监听
			pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
			// 启动缓存
			pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
			Thread.sleep(1000);

			// 创建3个子节点
			for (int i = 0; i < 3; i++) {
				instance.createNode(subWorkerPath + i, "创建节点" + i);
			}

			// 更新三个子节点
			for (int i = 0; i < 3; i++) {
				client.setData().forPath(subWorkerPath + i, ("第" + i + "个子节点数据").getBytes());
			}

			// 删除三个子节点
			for (int i = 0; i < 3; i++) {
				instance.deleteNode(subWorkerPath + i);
			}
		} catch (Exception e) {
			log.error("添加PathChildrenCache子节点事件失败，workerPath = {}", workerPath);
		}
	}

	/**
	 * 通过TreeCache实现事件监听
	 */
	public void treeCache() {
		ZKclient instance = ZKclient.instance;
		CuratorFramework client = instance.getClient();
		try {
			boolean nodeExist = instance.isNodeExist(workerPath);
			if (!nodeExist) {
				instance.createNode(workerPath, null);
			}

			// 创建监听事件
			TreeCacheListener treeCacheListener = new TreeCacheListener() {
				@Override
				public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
					ChildData data = event.getData();
					if (data == null) {
						log.error("节点数据为空");
						return;
					}
					String path = data.getPath();
					byte[] bytes = data.getData();
					switch (event.getType()) {
						case NODE_ADDED:
							// 节点添加事件
							log.info("[TreeCache]节点添加, path = {}, data = {}", path, new String(bytes));
							break;
						case NODE_UPDATED:
							// 节点更新事件
							log.info("[TreeCache]节点更新，path = {}, data = {}", path, new String(bytes));
							break;
						case NODE_REMOVED:
							// 节点删除事件
							log.info("[TreeCache]节点删除，path = {}, data = {}", path, new String(bytes));
							break;
						default:
							break;
					}
				}
			};

			// 构建PathChildrenCache实力
			TreeCache treeCache = new TreeCache(client, workerPath);

			// 添加事件监听
			treeCache.getListenable().addListener(treeCacheListener);
			// 启动缓存
			treeCache.start();
			Thread.sleep(1000);

			// 创建3个子节点
			for (int i = 0; i < 3; i++) {
				instance.createNode(subWorkerPath + i, "创建节点" + i);
			}
			Thread.sleep(1000);

			// 更新三个子节点
			for (int i = 0; i < 3; i++) {
				client.setData().forPath(subWorkerPath + i, ("第" + i + "个子节点数据").getBytes());
			}
			Thread.sleep(10000);

			// 删除三个子节点
			for (int i = 0; i < 3; i++) {
				instance.deleteNode(subWorkerPath + i);
			}
			Thread.sleep(1000);

			// 删除当前节点
			instance.deleteNode(workerPath);

			Thread.sleep(Integer.MAX_VALUE);
		} catch (Exception e) {
			log.error("添加TreeCache子节点事件失败，workerPath = {}", workerPath);
		}
	}
}
